package com.gis507.mapreduce.done;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.junit.Test;

import java.io.IOException;

public class AISClassMapper extends Mapper<LongWritable, Text,Text, AisClassifyBean> {

    Text outK = new Text();
    AisClassifyBean outV = new AisClassifyBean();
    private int sum = 1;
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        if (sum == 1){

            sum += 1;

        }else {
            try {
                String line = value.toString();

                String[] comment = line.split(",");

                //LAURA,V7JP9,9700665,538006090,货船,Cargo ship,在航,Under way using engine,
                // 199,32,11.8,179.0,179.6,11.0,108°0.354′E,19°3.948′N,0,MUMBAI,11-03 03:00,
                // 2020-01-01 08:10:26,1577837426,108.005900,19.065800,岸基

                String mmsi = comment[3];
                String shipType = comment[5];
                String nacStatusCN = comment[7];
                double length = UtilsClass.turnToDouble(comment[8]);
                double width = UtilsClass.turnToDouble(comment[9]);
                double draught = UtilsClass.turnToDouble(comment[10]);
                double heading = UtilsClass.turnToDouble(comment[11]);
                double course = UtilsClass.turnToDouble(comment[12]);
                double speed = UtilsClass.turnToDouble(comment[13]);
                String dest = comment[17];
                long unixTime = UtilsClass.turnToLong(comment[20]);
                double lon_d = UtilsClass.turnToDouble(comment[21]);
                double lat_d = UtilsClass.turnToDouble(comment[22]);
                //调用getTonnage方法算出吃水
                double tonnage = UtilsClass.getTonnage(shipType, length, width, draught);

                //fengzhuang
                outK.set(mmsi);

                outV.setMmsi(mmsi);
                outV.setShipType(shipType);
                outV.setNacStatusCN(nacStatusCN);
                outV.setLength(length);
                outV.setWidth(width);
                outV.setDraught(draught);
                outV.setHeading(heading);
                outV.setCourse(course);
                outV.setSpeed(speed);
                outV.setDest(dest);
                outV.setUnixTime(unixTime);
                outV.setLon_d(lon_d);
                outV.setLat_d(lat_d);
                outV.setTonnge(tonnage);

                //get the value of seaRange
                int seaRange = UtilsClass.getSeaRangeNumber(outV);

                //set seaRange for outV
                outV.setSeaRange(seaRange);

                //write for Mapper
                context.write(outK,outV);
            } catch (Exception e) {
                System.out.println("error");
                e.printStackTrace();
            }
        }
    }

    @Test
    public void test(){
//        String str1 = "";
//        String str2 = "23";
//
//        System.out.println(UtilsClass.turnToDouble(str1));
//        System.out.println(UtilsClass.turnToLong(str2));

        AisClassifyBean ais = new AisClassifyBean();
        ais.setLat_d(35.665205);
        ais.setLon_d(121.008173);

        AisClassifyBean newAis = UtilsClass.getNewAisClassifyBean(ais);

        System.out.println(ais + "---->" + newAis);
        System.out.println(ais.equals(newAis));

        int seaRangeNumber = UtilsClass.getSeaRangeNumber(ais);
        System.out.println(seaRangeNumber);
    }
}
